package cn.getech.data.development.service.impl;

import cn.getech.data.development.config.properties.BdpJobConfig;
import cn.getech.data.development.config.properties.ClouderaMangerConfig;
import cn.getech.data.development.entity.RealtimeQueueEntity;
import cn.getech.data.development.mapper.*;
import cn.getech.data.development.model.dto.CMClusterAnalyDto;
import cn.getech.data.development.model.dto.CMClusterDto;
import cn.getech.data.development.model.dto.JobRunTOPDto;
import cn.getech.data.development.model.dto.QueueStaticUsedHourDto;
import cn.getech.data.development.model.qo.CMClusterSearchQO;
import cn.getech.data.development.model.qo.CMClusterSearchTimeQO;
import cn.getech.data.development.model.vo.CMClusterCPUVO;
import cn.getech.data.development.model.vo.CMClusterHDFSVO;
import cn.getech.data.development.model.vo.CMClusterNetIOVO;
import cn.getech.data.development.model.vo.QueueStaticUsedHourVO;
import cn.getech.data.development.service.BigScreenService;
import cn.getech.data.development.utils.TimeUtil;
import cn.getech.data.development.utils.cloudera.CMApi;
import cn.getech.data.development.utils.cloudera.CMUtils;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.system.center.entity.SysTenantQueue;
import cn.getech.system.center.service.SysTenantQueueService;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.DecimalFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class BigScreenServiceImpl implements BigScreenService {

    @Autowired
    private ClouderaMangerConfig clouderaMangerConfig;
    @Autowired
    private BdpJobConfig bdpJobConfig;
    @Autowired
    private RealtimeQueueMapper realtimeQueueMapper;
    @Autowired
    private QueueStaticUsedHourMapper queueStaticUsedHourMapper;
    @Autowired
    private JobRunHistoryMapper jobRunHistoryMapper;
    @Autowired
    private JobStaticDayMapper jobStaticDayMapper;
    @Autowired
    private JobDepRunHistoryMapper jobDepRunHistoryMapper;
    @Autowired
    private SysTenantQueueService sysTenantQueueService;

    @Override
    public List<CMClusterAnalyDto> lastQueueMemory() {
        return lastQueueMemory(ShiroUtils.getTenantId());
    }

    @Override
    public List<CMClusterAnalyDto> lastQueueMemory(Integer tenantId) {
        CMClusterSearchTimeQO cmClusterSearchTimeQO = defaultCheackTime(null);
        List<CMClusterAnalyDto> cmClusterAnalyDtos = new ArrayList<>();
        try {
            String timeSeriesResponse = CMUtils.getTimeSeriesResponse(clouderaMangerConfig.getAddress()
                    , clouderaMangerConfig.getPort()
                    , clouderaMangerConfig.getUsername()
                    , clouderaMangerConfig.getPassword()
                    , CMApi.YARN_QUEUE_LAST_SQL
                    , cmClusterSearchTimeQO.getStartTime()
                    , cmClusterSearchTimeQO.getEndTime());
            if(StringUtils.isNotEmpty(timeSeriesResponse)){
                //查询对应的数据
                JSONArray jsonArray = analySizeData(timeSeriesResponse);
                if(CollectionUtil.isNotEmpty(jsonArray)){
                    for(int i = 0; i<jsonArray.size(); i++){
                        List<CMClusterAnalyDto> cmClusterAnalyDtos2 = analyTimeSeriesData(jsonArray, i);
                        if(CollectionUtil.isNotEmpty(cmClusterAnalyDtos2)){
                            cmClusterAnalyDtos.addAll(cmClusterAnalyDtos2);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("请求队列的实时数据异常！error:{}",e.getMessage());
            e.printStackTrace();
            throw new RRException("请求队列的实时数据异常！");
        }
        //查询对应的队列数据
        List<RealtimeQueueEntity> realtimeQueueEntities = realtimeQueueMapper.selectList(new QueryWrapper<RealtimeQueueEntity>());
        if(CollectionUtil.isNotEmpty(realtimeQueueEntities)){
            List<Long> list = null;
            //租户id
            if(null != tenantId){
                List<SysTenantQueue> list1 = sysTenantQueueService.list(new LambdaQueryWrapper<SysTenantQueue>().eq(SysTenantQueue::getTenantId, tenantId));
                if(CollectionUtil.isEmpty(list1)){
                    return new ArrayList<>();
                }
                list = new ArrayList<>();
                for (SysTenantQueue sysTenantQueue : list1) {
                    if(null != sysTenantQueue.getQueueId()){
                        list.add(Long.valueOf(sysTenantQueue.getQueueId()));
                    }
                }
            }
            List<CMClusterAnalyDto> restList = new ArrayList<>();
            for (RealtimeQueueEntity realtimeQueueEntity : realtimeQueueEntities) {
                if(!CollectionUtil.isEmpty(list)){
                    if(!list.contains(realtimeQueueEntity.getQueueId())){
                        continue;//没有则过滤掉
                    }
                }
                CMClusterAnalyDto cmClusterAnalyDto = new CMClusterAnalyDto();
                cmClusterAnalyDto.setQueueName(realtimeQueueEntity.getQueueName());
                cmClusterAnalyDto.setTotalMemoryMb(realtimeQueueEntity.getTotalMemoryMb()/1024);
                if(CollectionUtil.isNotEmpty(cmClusterAnalyDtos)){
                    CMClusterAnalyDto cmClusterAnalyDto1 = cmClusterAnalyDtos.stream().filter(o -> Objects.equals(o.getQueueName(), realtimeQueueEntity.getQueueName())).findAny().orElse(null);
                    if(null != cmClusterAnalyDto1){
                        cmClusterAnalyDto.setX(cmClusterAnalyDto1.getX());
                        cmClusterAnalyDto.setY(cmClusterAnalyDto1.getY()/1024);
                    }
                }
                restList.add(cmClusterAnalyDto);
            }
            return restList;
        }
        return null;
    }

    @Override
    public CMClusterDto focusSum() {
        CMClusterDto cmClusterDto = new CMClusterDto();
        List<CMClusterDto> cmClusterDtos = focusInfo(null,0);
        if(CollectionUtil.isNotEmpty(cmClusterDtos)){
            changeHostsCMClusterDto(cmClusterDto,cmClusterDtos);
            Integer count = realtimeQueueMapper.selectCount(new QueryWrapper<RealtimeQueueEntity>());
            //租户id
            Integer tenantId = ShiroUtils.getTenantId();
            if(null != tenantId){
                List<SysTenantQueue> list1 = sysTenantQueueService.list(new LambdaQueryWrapper<SysTenantQueue>().eq(SysTenantQueue::getTenantId, tenantId));
                if(CollectionUtil.isEmpty(list1)){
                    count = 0;
                }else{
                    count = list1.size();
                }
            }
            cmClusterDto.setQueueNum(count);
        }
        return cmClusterDto;
    }

    private void changeHostsCMClusterDto(CMClusterDto cmClusterDto, List<CMClusterDto> cmClusterDtos) {
        if(CollectionUtil.isNotEmpty(cmClusterDtos)){
            Double diskTotalSum = 0D;
            Double diskUsedSum = 0D;
            Double physicalMemoryTotalSum = 0D;
            Double physicalMemoryUsedSum = 0D;
            Integer sumCoreNums = 0;
            for (CMClusterDto clusterDto : cmClusterDtos) {
                if(null != clusterDto.getNumCores()){
                    sumCoreNums = sumCoreNums + clusterDto.getNumCores();
                }
                if(null != clusterDto.getDiskTotal()){
                    diskTotalSum = diskTotalSum + clusterDto.getDiskTotal();
                }
                if(null != clusterDto.getDiskUsed()){
                    diskUsedSum = diskUsedSum + clusterDto.getDiskUsed();
                }
                if(null != clusterDto.getPhysicalMemoryTotal()){
                    physicalMemoryTotalSum = physicalMemoryTotalSum + clusterDto.getPhysicalMemoryTotal();
                }
                if(null != clusterDto.getPhysicalMemoryUsed()){
                    physicalMemoryUsedSum = physicalMemoryUsedSum + clusterDto.getPhysicalMemoryUsed();
                }
            }
            cmClusterDto.setClusterNum(cmClusterDtos.size());
            if(0D != diskTotalSum){
                cmClusterDto.setDiskPercentageUsedSum(diskUsedSum/diskTotalSum * 100);
            }
            cmClusterDto.setDiskTotalSum(diskTotalSum);
            cmClusterDto.setDiskUsedSum(diskUsedSum);
            if(0D != physicalMemoryTotalSum){
                cmClusterDto.setPhysicalMemoryPercentageUsedSum(physicalMemoryUsedSum/physicalMemoryTotalSum * 100);
            }
            cmClusterDto.setPhysicalMemoryTotalSum(physicalMemoryTotalSum);
            cmClusterDto.setPhysicalMemoryUsedSum(physicalMemoryUsedSum);
            cmClusterDto.setNumCores(sumCoreNums);
        }
    }

    @Override
    public List<CMClusterDto> focusInfo(CMClusterSearchQO cmClusterSearchQO,Integer type) {
        //查询到所有的集群信息集合
        String loginUrl = CMApi.pingHttpUrl(clouderaMangerConfig.getAddress(),clouderaMangerConfig.getPort())
                + CMApi.LOGIN_API;
        String hostsUrl = CMApi.pingHttpUrl(clouderaMangerConfig.getAddress(),clouderaMangerConfig.getPort())
                + CMApi.HOST_OWERVIEW_API;
        String rest = CMUtils.getHttpData(loginUrl
                , hostsUrl
                , "j_username"
                , "j_password"
                , clouderaMangerConfig.getUsername()
                , clouderaMangerConfig.getPassword());
        if(StringUtils.isNotEmpty(rest)){
            List<CMClusterDto> cmClusterDtos = analyHostsJson(rest);
            if(CollectionUtil.isNotEmpty(cmClusterDtos)){
                if(Objects.equals(1,type)){
                    //遍历并且获取对应的cpu使用率
                    for (CMClusterDto cmClusterDto : cmClusterDtos) {
                        List<CMClusterAnalyDto> cmClusterAnalyDtos = clusterLastHostCPU(null, cmClusterDto.getHostId());
                        if(CollectionUtil.isNotEmpty(cmClusterAnalyDtos)){
                            cmClusterDto.setCpuPercentAcross(cmClusterAnalyDtos.get(0).getY());
                        }
                        //io读写
                        List<CMClusterAnalyDto> cmClusterAnalyDtos1 = clusterLastHostIOCPU(null, cmClusterDto.getHostId());
                        if(CollectionUtil.isNotEmpty(cmClusterAnalyDtos1)){
                            if(cmClusterAnalyDtos1.size() > 1){
                                cmClusterDto.setTotalReadIosRate(cmClusterAnalyDtos1.get(0).getY());
                                cmClusterDto.setTotalWriteIosRate(cmClusterAnalyDtos1.get(1).getY());
                            }else if(cmClusterAnalyDtos1.size() == 1){
                                cmClusterDto.setTotalReadIosRate(cmClusterAnalyDtos1.get(0).getY());
                            }
                        }
                    }
                }
                //转换单位
                for (CMClusterDto cmClusterDto : cmClusterDtos) {
                    cmClusterDto.setDiskUsed(cmClusterDto.getDiskUsed()/1024/1024/1024);
                    cmClusterDto.setDiskTotal(cmClusterDto.getDiskTotal()/1024/1024/1024);
                    cmClusterDto.setPhysicalMemoryUsed(cmClusterDto.getPhysicalMemoryUsed()/1024/1024/1024);
                    cmClusterDto.setPhysicalMemoryTotal(cmClusterDto.getPhysicalMemoryTotal()/1024/1024/1024);
                }
                if(null != cmClusterSearchQO){
                    String clusterName = cmClusterSearchQO.getClusterName();
                    String hostName = cmClusterSearchQO.getHostName();
                    if(StringUtils.isNotEmpty(clusterName) && StringUtils.isNotEmpty(hostName)){
                        return  cmClusterDtos.stream().filter(o -> (o.getClusterName().contains(clusterName)
                                && o.getHostName().contains(hostName))).collect(Collectors.toList());
                    }else if(StringUtils.isNotEmpty(clusterName)){
                        return  cmClusterDtos.stream().filter(o -> o.getClusterName().contains(clusterName)).collect(Collectors.toList());
                    }else if(StringUtils.isNotEmpty(hostName)){
                        return  cmClusterDtos.stream().filter(o -> o.getHostName().contains(hostName)).collect(Collectors.toList());
                    }
                }
                return cmClusterDtos;
            }
        }
        return null;
    }

    @Override
    public List<CMClusterAnalyDto> clusterCPU(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        //如果没有时间，则默认显示最近半个小时的数据
        cmClusterSearchTimeQO = defaultCheackTime(cmClusterSearchTimeQO);
        try {
            String timeSeriesResponse = CMUtils.getTimeSeriesResponse(clouderaMangerConfig.getAddress()
                    , clouderaMangerConfig.getPort()
                    , clouderaMangerConfig.getUsername()
                    , clouderaMangerConfig.getPassword()
                    , CMApi.CPU_HISTORY_SQL
                    , cmClusterSearchTimeQO.getStartTime()
                    , cmClusterSearchTimeQO.getEndTime());
            if(StringUtils.isNotEmpty(timeSeriesResponse)){
                JSONArray jsonArray = analySizeData(timeSeriesResponse);
                if(CollectionUtil.isNotEmpty(jsonArray)){
                    List<CMClusterAnalyDto> rest = analyTimeSeriesData(jsonArray,0);
                    if(CollectionUtil.isNotEmpty(rest)){
                        return rest;
                    }
                }
            }
        } catch (Exception e) {
            log.error("请求cpu数据异常！error:{}",e.getMessage());
            e.printStackTrace();
            throw new RRException("请求cpu数据异常！");
        }
        return null;
    }

    @Override
    public List<CMClusterAnalyDto> clusterLastHostCPU(CMClusterSearchTimeQO cmClusterSearchTimeQO,String hostId) {
        //如果没有时间，则默认显示最近半个小时的数据
        cmClusterSearchTimeQO = defaultCheackTime(cmClusterSearchTimeQO);
        try {
            String timeSeriesResponse = CMUtils.getTimeSeriesResponse(clouderaMangerConfig.getAddress()
                    , clouderaMangerConfig.getPort()
                    , clouderaMangerConfig.getUsername()
                    , clouderaMangerConfig.getPassword()
                    , CMApi.LAST_HOST_CPU_HISTORY_SQL + hostId
                    , cmClusterSearchTimeQO.getStartTime()
                    , cmClusterSearchTimeQO.getEndTime());
            if(StringUtils.isNotEmpty(timeSeriesResponse)){
                JSONArray jsonArray = analySizeData(timeSeriesResponse);
                if(CollectionUtil.isNotEmpty(jsonArray)){
                    List<CMClusterAnalyDto> rest = analyTimeSeriesData(jsonArray,0);
                    if(CollectionUtil.isNotEmpty(rest)){
                        return rest;
                    }
                }
            }
        } catch (Exception e) {
            log.error("请求cpu单个节点的使用率数据异常！error:{}",e.getMessage());
            e.printStackTrace();
            throw new RRException("请求cpu单个节点的使用率数据异常！");
        }
        return null;
    }

    @Override
    public List<CMClusterAnalyDto> clusterLastHostIOCPU(CMClusterSearchTimeQO cmClusterSearchTimeQO, String hostId) {
        //如果没有时间，则默认显示最近半个小时的数据
        cmClusterSearchTimeQO = defaultCheackTime(cmClusterSearchTimeQO);
        try {
            String timeSeriesResponse = CMUtils.getTimeSeriesResponse(clouderaMangerConfig.getAddress()
                    , clouderaMangerConfig.getPort()
                    , clouderaMangerConfig.getUsername()
                    , clouderaMangerConfig.getPassword()
                    , CMApi.LAST_HOST_IO_CPU_HISTORY_SQL + hostId
                    , cmClusterSearchTimeQO.getStartTime()
                    , cmClusterSearchTimeQO.getEndTime());
            if(StringUtils.isNotEmpty(timeSeriesResponse)){
                JSONArray jsonArray = analySizeData(timeSeriesResponse);
                if(CollectionUtil.isNotEmpty(jsonArray)){
                    List<CMClusterAnalyDto> allLists = new ArrayList<>();
                    List<CMClusterAnalyDto> rest = analyTimeSeriesData(jsonArray,0);
                    if(CollectionUtil.isNotEmpty(rest)){
                        allLists.add(rest.get(0));
                    }
                    List<CMClusterAnalyDto> rest1 = analyTimeSeriesData(jsonArray,1);
                    if(CollectionUtil.isNotEmpty(rest1)){
                        allLists.add(rest1.get(0));
                    }
                    return allLists;
                }
            }
        } catch (Exception e) {
            log.error("请求cpu单个节点的io读写数据异常！error:{}",e.getMessage());
            e.printStackTrace();
            throw new RRException("请求cpu单个节点的io读写数据异常！");
        }
        return null;
    }

    @Override
    public CMClusterNetIOVO clusterNetIO(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        //如果没有时间，则默认显示最近半个小时的数据
        cmClusterSearchTimeQO = defaultCheackTime(cmClusterSearchTimeQO);
        try {
            String timeSeriesResponse = CMUtils.getTimeSeriesResponse(clouderaMangerConfig.getAddress()
                    , clouderaMangerConfig.getPort()
                    , clouderaMangerConfig.getUsername()
                    , clouderaMangerConfig.getPassword()
                    , CMApi.NET_IO_HISTORY_SQL
                    , cmClusterSearchTimeQO.getStartTime()
                    , cmClusterSearchTimeQO.getEndTime());
            if(StringUtils.isNotEmpty(timeSeriesResponse)){
                CMClusterNetIOVO cmClusterNetIOVO = new CMClusterNetIOVO();
                JSONArray jsonArray = analySizeData(timeSeriesResponse);
                if(CollectionUtil.isNotEmpty(jsonArray)){
                    List<CMClusterAnalyDto> cmClusterAnalyDtos = analyTimeSeriesData(jsonArray,0);
                    if(CollectionUtil.isNotEmpty(cmClusterAnalyDtos)){
                        for (CMClusterAnalyDto cmClusterAnalyDto : cmClusterAnalyDtos) {
                            changeCMType(cmClusterAnalyDto,cmClusterSearchTimeQO.getCmType());
                        }
                        cmClusterNetIOVO.setClusters1(changeTimeData(cmClusterAnalyDtos));
                    }

                    if(jsonArray.size() > 1){
                        List<CMClusterAnalyDto> cmClusterAnalyDtos1 = analyTimeSeriesData(jsonArray, 1);
                        if(CollectionUtil.isNotEmpty(cmClusterAnalyDtos1)){
                            for (CMClusterAnalyDto cmClusterAnalyDto : cmClusterAnalyDtos1) {
                                changeCMType(cmClusterAnalyDto,cmClusterSearchTimeQO.getCmType());
                            }
                            cmClusterNetIOVO.setClusters2(changeTimeData(cmClusterAnalyDtos1));
                        }

                    }
                    return cmClusterNetIOVO;
                }
            }
        } catch (Exception e) {
            log.error("请求网络io数据异常！error:{}",e.getMessage());
            e.printStackTrace();
            throw new RRException("请求网络io数据异常！");
        }
        return null;
    }

    @Override
    public CMClusterHDFSVO clusterHdfsInfo(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        //如果没有时间，则默认显示最近半个小时的数据
        cmClusterSearchTimeQO = defaultCheackTime(cmClusterSearchTimeQO);
        try {
            String timeSeriesResponse = CMUtils.getTimeSeriesResponse(clouderaMangerConfig.getAddress()
                    , clouderaMangerConfig.getPort()
                    , clouderaMangerConfig.getUsername()
                    , clouderaMangerConfig.getPassword()
                    , CMApi.HDFS_HISTORY_SQL + bdpJobConfig.getNamespace()
                    , cmClusterSearchTimeQO.getStartTime()
                    , cmClusterSearchTimeQO.getEndTime());
            if(StringUtils.isNotEmpty(timeSeriesResponse)){
                CMClusterHDFSVO cmClusterHDFSVO = new CMClusterHDFSVO();
                JSONArray jsonArray = analySizeData(timeSeriesResponse);
                if(CollectionUtil.isNotEmpty(jsonArray)){
                    List<CMClusterAnalyDto> totalHdfsList = analyTimeSeriesData(jsonArray, 0);
                    //改变单位
                    if(CollectionUtil.isNotEmpty(totalHdfsList)){
                        for (CMClusterAnalyDto cmClusterAnalyDto : totalHdfsList) {
                            changeCMType(cmClusterAnalyDto,cmClusterSearchTimeQO.getCmType());
                        }
                        cmClusterHDFSVO.setClusters1(changeTimeData(totalHdfsList));
                    }
                    if(jsonArray.size() > 1){
                        List<CMClusterAnalyDto> useHdfsList = analyTimeSeriesData(jsonArray, 1);
                        if(CollectionUtil.isNotEmpty(useHdfsList)){
                            for (CMClusterAnalyDto cmClusterAnalyDto : useHdfsList) {
                                changeCMType(cmClusterAnalyDto,cmClusterSearchTimeQO.getCmType());
                            }
                            cmClusterHDFSVO.setClusters2(changeTimeData(useHdfsList));
                        }
                    }
                    if(jsonArray.size() > 2){
                        List<CMClusterAnalyDto> useHdfsList = analyTimeSeriesData(jsonArray, 2);
                        if(CollectionUtil.isNotEmpty(useHdfsList)){
                            for (CMClusterAnalyDto cmClusterAnalyDto : useHdfsList) {
                                changeCMType(cmClusterAnalyDto,cmClusterSearchTimeQO.getCmType());
                            }
                            cmClusterHDFSVO.setClusters3(changeTimeData(useHdfsList));
                        }
                    }
                    return cmClusterHDFSVO;
                }
            }
        } catch (Exception e) {
            log.error("请求hdfs数据异常！error:{}",e.getMessage());
            e.printStackTrace();
            throw new RRException("请求hdfs数据异常！");
        }
        return null;
    }

    @Override
    public List<CMClusterCPUVO> changeTimeData(List<CMClusterAnalyDto> totalHdfsList) {
        if(CollectionUtil.isNotEmpty(totalHdfsList)){
            List<CMClusterCPUVO> lists = new ArrayList<>();
            for (CMClusterAnalyDto cmClusterAnalyDto : totalHdfsList) {
                CMClusterCPUVO cmClusterCPUVO = new CMClusterCPUVO();
                BeanUtils.copyProperties(cmClusterAnalyDto,cmClusterCPUVO);
                //去掉秒
                String s = TimeUtil.changeDateTime(cmClusterAnalyDto.getX());
                if(StringUtils.isNotEmpty(s) && s.length() > 16){
                    cmClusterCPUVO.setX(s.substring(5,16));
                }
                lists.add(cmClusterCPUVO);
            }
            return lists;
        }
        return null;
    }

    private void changeCMType(CMClusterAnalyDto cmClusterAnalyDto, Integer cmType) {
        //单位类型： 0：T  1:G  2:M  3:K/S  4:M/S  5:G/S
        if(Objects.equals(0,cmType)){
            Double y = cmClusterAnalyDto.getY();
            if(null != y){
                cmClusterAnalyDto.setY(y/1024/1024/1024/1024);
            }
        }
        if(Objects.equals(1,cmType)){
            Double y = cmClusterAnalyDto.getY();
            if(null != y){
                cmClusterAnalyDto.setY(y/1024/1024/1024);
            }
        }
        if(Objects.equals(2,cmType)){
            Double y = cmClusterAnalyDto.getY();
            if(null != y){
                cmClusterAnalyDto.setY(y/1024/1024);
            }
        }

        if(Objects.equals(3,cmType)){
            Double y = cmClusterAnalyDto.getY();
            if(null != y){
                cmClusterAnalyDto.setY(y/1024);
            }
        }

        if(Objects.equals(4,cmType)){
            Double y = cmClusterAnalyDto.getY();
            if(null != y){
                cmClusterAnalyDto.setY(y/1024/1024);
            }
        }

        if(Objects.equals(5,cmType)){
            Double y = cmClusterAnalyDto.getY();
            if(null != y){
                cmClusterAnalyDto.setY(y/1024/1024/1024);
            }
        }
    }

    @Override
    public List<QueueStaticUsedHourDto> queueMemoryHistory(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        cheackTimeData(cmClusterSearchTimeQO);
        cmClusterSearchTimeQO.setTenantId(ShiroUtils.getTenantId());
        //查询到这个时间段的所有的历史数据
        return queueStaticUsedHourMapper.queryList(cmClusterSearchTimeQO);
    }

    private void cheackTimeData(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        LocalDateTime now = LocalDateTime.now();
        if(null == cmClusterSearchTimeQO){
            throw new RRException("请求类型必须传递！");
        }
        if(null == cmClusterSearchTimeQO.getType()){
            throw new RRException("请求类型必须传递！");
        }
        //今天
        if(0 == cmClusterSearchTimeQO.getType()){
            cmClusterSearchTimeQO.setStartTime(TimeUtil.changeStartDateTime(now));
            cmClusterSearchTimeQO.setEndTime(TimeUtil.changeEndDateTime(now));
        }
        //本周
        if(1 == cmClusterSearchTimeQO.getType()){
            cmClusterSearchTimeQO.setStartTime(TimeUtil.getWeekStart());
            String weekEnd = TimeUtil.getWeekEnd();
            String todayStart = TimeUtil.changeStartDateTime(now);
            LocalDateTime endDateTime = TimeUtil.cheackDateTime(weekEnd);
            LocalDateTime todayStartDateTime = TimeUtil.cheackDateTime(todayStart);
            if(endDateTime.isAfter(todayStartDateTime.plusDays(1))){
                cmClusterSearchTimeQO.setEndTime(TimeUtil.changeEndDateTime(now));
            }else {
                cmClusterSearchTimeQO.setEndTime(weekEnd);
            }
        }
        if(2 == cmClusterSearchTimeQO.getType()){
            //开始时间和结束时间必须填写
            if(StringUtils.isEmpty(cmClusterSearchTimeQO.getStartTime())){
                throw new RRException("开始时间必须传递！");
            }
            if(StringUtils.isEmpty(cmClusterSearchTimeQO.getEndTime())){
                throw new RRException("结束时间必须传递！");
            }
            //开始时间必须小于当前时间
            LocalDateTime startTime = TimeUtil.cheackDateTime(cmClusterSearchTimeQO.getStartTime());
            if(startTime.isAfter(now)){
                throw new RRException("开始时间必须小于当前时间！");
            }
            //结束时间如果大于当前时间，则使用当前时间
            LocalDateTime endTime = TimeUtil.cheackDateTime(cmClusterSearchTimeQO.getEndTime());
            if(endTime.isAfter(now)){
                cmClusterSearchTimeQO.setEndTime(TimeUtil.changeEndDateTime(now));
            }
        }
        //本月
        if(3 == cmClusterSearchTimeQO.getType()){
            cmClusterSearchTimeQO.setStartTime(TimeUtil.getMonthStart());
            String monthEnd = TimeUtil.getMonthEnd();
            String todayStart = TimeUtil.changeStartDateTime(now);
            LocalDateTime endDateTime = TimeUtil.cheackDateTime(monthEnd);
            LocalDateTime todayStartDateTime = TimeUtil.cheackDateTime(todayStart);
            if(endDateTime.isAfter(todayStartDateTime.plusDays(1))){
                cmClusterSearchTimeQO.setEndTime(TimeUtil.changeEndDateTime(now));
            }else {
                cmClusterSearchTimeQO.setEndTime(monthEnd);
            }
        }
    }

    @Override
    public QueueStaticUsedHourVO queueMemoryHistoryCharts(List<QueueStaticUsedHourDto> rest, CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        LocalDateTime now = LocalDateTime.now();
        if(CollectionUtil.isNotEmpty(rest)){
            DecimalFormat df = new DecimalFormat("0.00");
            Map<String, List<QueueStaticUsedHourDto>> collect = rest.stream().collect(Collectors.groupingBy(QueueStaticUsedHourDto::getQueueName));
            QueueStaticUsedHourVO queueStaticUsedHourVO = new QueueStaticUsedHourVO();
            List<String> sTimes = new ArrayList<>();
            if(cmClusterSearchTimeQO.getType() == 0){
                //以小时制返回
                int hour = now.getHour();
                for(int i = 1; i<=hour; i++){
                    sTimes.add(String.valueOf(i));
                }
            }else{
                //以天制返回
                List<String> timeOriginalList = TimeUtil.getTimeOriginalList(cmClusterSearchTimeQO.getStartTime(), cmClusterSearchTimeQO.getEndTime());
                sTimes.addAll(timeOriginalList);
            }
            Map<String, List<Double>> usedHourMaps = new HashMap<>();
            for(Map.Entry<String, List<QueueStaticUsedHourDto>> entry : collect.entrySet()){
                String key = entry.getKey();
                List<QueueStaticUsedHourDto> value = entry.getValue();
                List<Double> sumDataList = new ArrayList<>();
                for (String sTime : sTimes) {
                    if(cmClusterSearchTimeQO.getType() == 0){
                        if(CollectionUtil.isNotEmpty(value)){
                            QueueStaticUsedHourDto queueStaticUsedHourDto = value.stream().filter(o -> Objects.equals(sTime, String.valueOf(TimeUtil.cheackDateTime(o.getStaticTime().substring(0,19)).getHour()))).findAny().orElse(null);
                            if(null != queueStaticUsedHourDto){
                                sumDataList.add(queueStaticUsedHourDto.getStaticUsedPercent());
                                continue;
                            }
                        }
                        sumDataList.add(0D);
                    }else{
                        if(CollectionUtil.isNotEmpty(value)) {
                            List<QueueStaticUsedHourDto> queueStaticUsedHourDtoss = value.stream().filter(o -> Objects.equals(sTime, TimeUtil.changeDate(TimeUtil.cheackDateTime(o.getStaticTime().substring(0,19))))).collect(Collectors.toList());
                            if(CollectionUtil.isNotEmpty(queueStaticUsedHourDtoss)){
                                Double usedNum = queueStaticUsedHourDtoss.stream().collect(Collectors.averagingDouble(QueueStaticUsedHourDto::getStaticUsedPercent));
                                sumDataList.add(Double.valueOf(df.format(usedNum)));
                                continue;
                            }
                        }
                        sumDataList.add(0D);
                    }
                }
                usedHourMaps.put(key,sumDataList);
            }
            queueStaticUsedHourVO.setUsedHourMaps(usedHourMaps);
            queueStaticUsedHourVO.setXstaticTime(sTimes);
            return queueStaticUsedHourVO;
        }
        return null;
    }

    @Override
    public List<JobRunTOPDto> runTimeTop(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        cheackTimeData(cmClusterSearchTimeQO);
        //根据开始时间和结束时间查询
        cmClusterSearchTimeQO.setTenantId(ShiroUtils.getTenantId());
        List<JobRunTOPDto> jobRunTOPDtos = jobRunHistoryMapper.runTimeTop(cmClusterSearchTimeQO);
        List<JobRunTOPDto> jobDepRunTOPDtos = jobDepRunHistoryMapper.runTimeTop(cmClusterSearchTimeQO);
        List<JobRunTOPDto> allLists = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(jobRunTOPDtos)){
            allLists.addAll(jobRunTOPDtos);
        }
        if(CollectionUtil.isNotEmpty(jobDepRunTOPDtos)){
            allLists.addAll(jobDepRunTOPDtos);
        }
        if(CollectionUtil.isNotEmpty(allLists)){
            return allLists.stream().sorted(Comparator.comparing(JobRunTOPDto :: getDurTimes).reversed()).limit(20).collect(Collectors.toList());
        }
        return null;
    }

    @Override
    public List<JobRunTOPDto> runFailNumTop(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        cheackTimeData(cmClusterSearchTimeQO);
        cmClusterSearchTimeQO.setTenantId(ShiroUtils.getTenantId());
        return jobStaticDayMapper.runFailNumTop(cmClusterSearchTimeQO);
    }

    @Override
    public List<CMClusterAnalyDto> clusterLastCPU() {
        //如果没有时间，则默认显示最近半个小时的数据
        CMClusterSearchTimeQO cmClusterSearchTimeQO = defaultCheackTime(null);
        try {
            String timeSeriesResponse = CMUtils.getTimeSeriesResponse(clouderaMangerConfig.getAddress()
                    , clouderaMangerConfig.getPort()
                    , clouderaMangerConfig.getUsername()
                    , clouderaMangerConfig.getPassword()
                    , CMApi.LAST_CPU_HISTORY_SQL
                    , cmClusterSearchTimeQO.getStartTime()
                    , cmClusterSearchTimeQO.getEndTime());
            if(StringUtils.isNotEmpty(timeSeriesResponse)){
                JSONArray jsonArray = analySizeData(timeSeriesResponse);
                if(CollectionUtil.isNotEmpty(jsonArray)){
                    List<CMClusterAnalyDto> rest = analyTimeSeriesData(jsonArray,0);
                    if(CollectionUtil.isNotEmpty(rest)){
                        return rest;
                    }
                }
            }
        } catch (Exception e) {
            log.error("请求cpu最近的使用率数据异常！error:{}",e.getMessage());
            e.printStackTrace();
            throw new RRException("请求cpu最近的使用率数据异常！");
        }
        return null;
    }


    private List<CMClusterAnalyDto> analyClusterData(JSONObject jsonObj) {
        if(null != jsonObj){
            Object data = jsonObj.get("data");
            if(null != data){
                JSONArray rest = JSONArray.parseArray(data.toString());
                if(CollectionUtil.isNotEmpty(rest)){
                    //解析队列名字
                    String queueName = "";
                    Object metadata = jsonObj.get("metadata");
                    if(null != metadata){
                        JSONObject jsonObject = JSONObject.parseObject(metadata.toString());
                        if(null != jsonObject){
                            Object attributes = jsonObject.get("attributes");
                            if(null != attributes){
                                JSONObject obj = JSONObject.parseObject(attributes.toString());
                                if(null != obj){
                                    queueName = obj.getString("queueName");
                                }
                            }
                        }
                    }
                    List<CMClusterAnalyDto> restList = new ArrayList<>();
                    for (Object o : rest) {
                        if(null != o){
                            JSONObject jsonObject = JSONObject.parseObject(o.toString());
                            if(null != jsonObject){
                                CMClusterAnalyDto cmClusterAnalyDto = new CMClusterAnalyDto();
                                cmClusterAnalyDto.setQueueName(queueName);
                                String timestamp = jsonObject.getString("timestamp");
                                if(StringUtils.isNotEmpty(timestamp)){
                                    cmClusterAnalyDto.setX(CMUtils.transCMDateToNormal(timestamp));
                                }
                                cmClusterAnalyDto.setY(jsonObject.getDouble("value"));
                                Object rollupStatistics = jsonObject.get("aggregateStatistics");
                                if(null != rollupStatistics){
                                    JSONObject obj01 = JSONObject.parseObject(rollupStatistics.toString());
                                    if(null != obj01){
                                        cmClusterAnalyDto.setCount(obj01.getInteger("count"));
                                        cmClusterAnalyDto.setMax(obj01.getDouble("max"));
                                        cmClusterAnalyDto.setMin(obj01.getDouble("min"));
                                        cmClusterAnalyDto.setMean(obj01.getDouble("mean"));
                                        String maxTime = obj01.getString("maxTime");
                                        String minTime = obj01.getString("minTime");
                                        if(StringUtils.isNotEmpty(maxTime)){
                                            cmClusterAnalyDto.setMaxTimestampMs(CMUtils.transCMDateToNormal(maxTime));
                                        }
                                        if(StringUtils.isNotEmpty(minTime)){
                                            cmClusterAnalyDto.setMinTimestampMs(CMUtils.transCMDateToNormal(minTime));
                                        }
                                        cmClusterAnalyDto.setStdDev(obj01.getDouble("stdDev"));
                                        Object crossEntityMetadata = obj01.get("crossEntityMetadata");
                                        if(null != crossEntityMetadata){
                                            JSONObject obj02 = JSONObject.parseObject(crossEntityMetadata.toString());
                                            if(null != obj02){
                                                cmClusterAnalyDto.setMaxEntityName(obj02.getString("maxEntityDisplayName"));
                                                cmClusterAnalyDto.setMinEntityName(obj02.getString("minEntityDisplayName"));
                                            }
                                        }
                                    }
                                }
                                restList.add(cmClusterAnalyDto);
                            }
                        }
                    }
                    return restList;
                }
            }
        }
        return null;
    }

    private List<CMClusterAnalyDto> analyTimeSeriesData(JSONArray jsonArray,Integer index) {
        if(CollectionUtil.isNotEmpty(jsonArray)){
            if(jsonArray.size() > index){
                Object obj = jsonArray.get(index);
                if(null != obj){
                    JSONObject jsonObj = JSONObject.parseObject(obj.toString());
                    if(null != jsonObj){
                        return analyClusterData(jsonObj);
                    }
                }
            }else{
                return null;
            }
        }
        return null;
    }

    private JSONArray analySizeData(String timeSeriesResponse) {
        if(StringUtils.isNotEmpty(timeSeriesResponse)){
            JSONObject objRoot = JSONObject.parseObject(timeSeriesResponse);
            if(null != objRoot){
                Object items = objRoot.get("items");
                if(null != items){
                    JSONArray jsonArrayTest = JSONArray.parseArray(items.toString());
                    if(CollectionUtil.isNotEmpty(jsonArrayTest)){
                        Object o = jsonArrayTest.get(0);
                        if(null != o){
                            JSONObject jsonObject = JSONObject.parseObject(o.toString());
                            if(null != jsonObject){
                                Object timeSeries = jsonObject.get("timeSeries");
                                if(null != timeSeries){
//                                    log.info(timeSeries.toString());
                                    JSONArray jarr = JSONArray.parseArray(timeSeries.toString());
                                    if(CollectionUtil.isNotEmpty(jarr)){
                                        return jarr;
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
        return null;
    }

    private CMClusterSearchTimeQO defaultCheackTime(CMClusterSearchTimeQO cmClusterSearchTimeQO) {
        LocalDateTime now = LocalDateTime.now();
        if(null == cmClusterSearchTimeQO){
            cmClusterSearchTimeQO = new CMClusterSearchTimeQO();
            //设置默认的时间
            cmClusterSearchTimeQO.setEndTime(TimeUtil.changeLocalDateTime(now));
            LocalDateTime yesLocTime = now.plusMinutes(-30);
            cmClusterSearchTimeQO.setStartTime(TimeUtil.changeLocalDateTime(yesLocTime));

        } else if((StringUtils.isEmpty(cmClusterSearchTimeQO.getStartTime()) && StringUtils.isEmpty(cmClusterSearchTimeQO.getEndTime()))){
            //设置默认的时间
            cmClusterSearchTimeQO.setEndTime(TimeUtil.changeLocalDateTime(now));
            LocalDateTime yesLocTime = now.plusMinutes(-30);
            cmClusterSearchTimeQO.setStartTime(TimeUtil.changeLocalDateTime(yesLocTime));
        }else{
            if(StringUtils.isEmpty(cmClusterSearchTimeQO.getStartTime())){
                throw new RRException("开始时间必须传递！");
            }
            if(StringUtils.isEmpty(cmClusterSearchTimeQO.getEndTime())){
                throw new RRException("结束时间必须传递！");
            }
            //验证其数据格式
            try {
                TimeUtil.cheackDateTime(cmClusterSearchTimeQO.getStartTime());
            }catch (Exception e){
                throw new RRException("开始时间格式不正确！" + e.getMessage());
            }
            //验证其数据格式
            try {
                TimeUtil.cheackDateTime(cmClusterSearchTimeQO.getEndTime());
            }catch (Exception e){
                throw new RRException("结束时间格式不正确！" + e.getMessage());
            }
        }
        return cmClusterSearchTimeQO;
    }

    private List<CMClusterDto> analyHostsJson(String rest) {
        if(StringUtils.isNotEmpty(rest)){
            JSONObject jsonObject = JSONObject.parseObject(rest);
            if(null != jsonObject){
                Object hosts = jsonObject.get("hosts");
                if(null != hosts){
                    JSONArray jsonArray = JSONArray.parseArray(hosts.toString());
                    if(CollectionUtil.isNotEmpty(jsonArray)){
                        List<CMClusterDto> listDatas = new ArrayList<>();
                        for (Object o : jsonArray) {
                            if(null != o){
                                JSONObject obj = JSONObject.parseObject(o.toString());
                                if(null != obj){
                                    CMClusterDto cmClusterDto = new CMClusterDto();
                                    cmClusterDto.setCdhVersion(obj.getString("cdhVersion"));
                                    cmClusterDto.setClusterName(obj.getString("clusterName"));
                                    cmClusterDto.setHostId(obj.getString("hostId"));
                                    cmClusterDto.setHostName(obj.getString("hostName"));
                                    cmClusterDto.setNumCores(obj.getInteger("numCores"));
                                    cmClusterDto.setPhysicalMemoryPercentageUsed(obj.getDouble("physicalMemoryPercentageUsed"));
                                    cmClusterDto.setPhysicalMemoryTotal(obj.getDouble("physicalMemoryTotal"));
                                    cmClusterDto.setPhysicalMemoryUsed(obj.getDouble("physicalMemoryUsed"));
                                    cmClusterDto.setDiskPercentageUsed(obj.getDouble("diskPercentageUsed"));
                                    cmClusterDto.setDiskTotal(obj.getDouble("diskTotal"));
                                    cmClusterDto.setDiskUsed(obj.getDouble("diskUsed"));
                                    listDatas.add(cmClusterDto);
                                }
                            }
                        }
                        return listDatas;
                    }
                }
            }
        }
        return null;
    }
}
